package com.atguigu.udmp.stream.common.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Properties;

public class MyKafkaUtil {
    private static String BOOTSTRAP_SERVERS = MyPropertiesUtil.load("config.properties").getProperty("kafka.server");
    private static String DEFAULT_TOPIC = "default_topic";


    public static FlinkKafkaConsumer<String> getKafkaConsumer(List<String> topicList, String groupId) {
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                topicList,  new SimpleStringSchema(),prop);
        return consumer;
    }



    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60 * 15 * 1000 + "");  //要考虑checkpoint时间

        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(
                DEFAULT_TOPIC,
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
                        return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
                    }
                },
                prop,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        return producer;
    }
}

